package com.atguigu.gmall.realtime.util;

import org.apache.flink.api.common.serialization.DeserializationSchema;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.connector.kafka.sink.KafkaRecordSerializationSchema;
import org.apache.flink.connector.kafka.sink.KafkaSink;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import java.io.IOException;

/**
 * @author Felix
 * @date 2023/8/19
 * 操作kafka的工具类
 */
public class MyKafkaUtil {
    private static final String KAFKA_SERVER = "hadoop102:9092,hadoop103:9092,hadoop104:9092";

    //获取KafkaSource
    public static KafkaSource<String> getKafkaSource(String topic, String groupId) {
        KafkaSource<String> kafkaSource = KafkaSource.<String>builder()
            .setBootstrapServers(KAFKA_SERVER)
            .setTopics(topic)
            .setGroupId(groupId)
            //在生产环境中，为了保证一致性，我们在消费数据的时候，预提交的数据不应该消费
            .setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed")
            //在生产环境中，从flink维度的偏移量位置开始消费，如果没有找到从指定位置进行消费
            // .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.LATEST))
            //在学习阶段，我们直接从kafka的主题最新偏移量位置开始消费
            .setStartingOffsets(OffsetsInitializer.latest())
            //默认情况下，使用SimpleStringSchema进行反序列化，如果读取的消息为空，处理不了
            // .setValueOnlyDeserializer(new SimpleStringSchema())
            .setValueOnlyDeserializer(
                new DeserializationSchema<String>() {
                    @Override
                    public String deserialize(byte[] message) throws IOException {
                        if (message != null) {
                            return new String(message);
                        }
                        return null;
                    }

                    @Override
                    public boolean isEndOfStream(String nextElement) {
                        return false;
                    }

                    @Override
                    public TypeInformation<String> getProducedType() {
                        return TypeInformation.of(String.class);
                    }
                }
            )
            .build();
        return kafkaSource;
    }

    //获取kafkaSink
    public static KafkaSink<String> getKafkaSink(String topic) {
        KafkaSink<String> kafkaSink = KafkaSink.<String>builder()
            .setBootstrapServers(KAFKA_SERVER)
            .setRecordSerializer(KafkaRecordSerializationSchema.builder()
                .setTopic(topic)
                .setValueSerializationSchema(new SimpleStringSchema())
                .build()
            )
            //在生产环境中，如果要想保证写入的一致性，需要进行如下的日志
            //1.指定DeliveryGuarantee.EXACTLY_ONCE
            //2.setTransactionalIdPrefix("")
            //3.检查点开启
            //4.事务超时时间大于检查点超时时间  但是要小于事务最大超时时间(默认15min)
            //5.在消费端，对于没有提交的消息，不应该被消费
            // .setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed")
            // .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
            // .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,15*60*1000 + "")
            .build();
        return kafkaSink;
    }

    //获取从topic_db主题中读取数据 创建动态表的建表语句
    public static String getTopicDbDDL(String groupId) {
        return "CREATE TABLE topic_db (\n" +
            "  `database` string,\n" +
            "  `table` string,\n" +
            "  `type` string,\n" +
            "  `ts` string,\n" +
            "  `data` map<string,string>,\n" +
            "  `old` map<string,string>,\n" +
            "  proc_time AS PROCTIME()\n" +
            ") " + getKafkaDDL("topic_db", groupId);
    }

    //获取kafka连接器连接选项
    public static String getKafkaDDL(String topic, String groupId) {
        return "WITH (\n" +
            "  'connector' = 'kafka',\n" +
            "  'topic' = '" + topic + "',\n" +
            "  'properties.bootstrap.servers' = '" + KAFKA_SERVER + "',\n" +
            "  'properties.group.id' = '" + groupId + "',\n" +
            // "  //在生产环境中，偏移量位置应该从程序维护的位置读取,如果在flink中没有找到维护的偏移量，从kafka最新位置开始消费\n" +
            // "  //'scan.startup.mode' = 'group-offsets',\n" +
            // "  //'properties.auto.offset.reset' = 'latest',\n" +
            "  'scan.startup.mode' = 'latest-offset',\n" +
            "  'format' = 'json'\n" +
            ")";
    }

    //获取upsert-kafka连接器连接选项
    public static String getUpsertKafkaDDL(String topic) {
        return "WITH (\n" +
            "  'connector' = 'upsert-kafka',\n" +
            "  'topic' = '" + topic + "',\n" +
            "  'properties.bootstrap.servers' = '" + KAFKA_SERVER + "',\n" +
            "  'key.format' = 'json',\n" +
            "  'value.format' = 'json'\n" +
            ")";
    }

    //获取kafkaSink
    public static <T>KafkaSink<T> getKafkaSinkBySchema(KafkaRecordSerializationSchema<T> krs) {
        KafkaSink<T> kafkaSink = KafkaSink.<T>builder()
            .setBootstrapServers(KAFKA_SERVER)
            .setRecordSerializer(krs)
            //在生产环境中，如果要想保证写入的一致性，需要进行如下的日志
            //1.指定DeliveryGuarantee.EXACTLY_ONCE
            //2.setTransactionalIdPrefix("")
            //3.检查点开启
            //4.事务超时时间大于检查点超时时间  但是要小于事务最大超时时间(默认15min)
            //5.在消费端，对于没有提交的消息，不应该被消费
            // .setProperty(ConsumerConfig.ISOLATION_LEVEL_CONFIG,"read_committed")
            // .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
            // .setProperty(ProducerConfig.TRANSACTION_TIMEOUT_CONFIG,15*60*1000 + "")
            .build();
        return kafkaSink;
    }
}
